package com.shujia.jinjie;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 *  1. 如何自定义拦截器?
 *   flume的自定义拦截器需要实现Flume提供的Interceptor接口.
 *
 *  实现抽象方法:
 *      initialize: 完成一些初始化工作.
 *      close: 完成一些善后的工作
 *      intercept：拦截器的核心处理方法.  拦截的逻辑.
 *          intercept(Event event) : 单个event的拦截处理
 *          intercept(List<Event> events): 批次event的拦截处理
 *
 *  2. 拦截器的对象如何实例化?
 *    在拦截器中定义一个static的内部类，实现Flume提供的Builder接口
 *
 *   实现抽象方法:
 *      build : 用于构建拦截器对象
 *      configure：用于读取配置信息（xxxx.conf）
 *
 *
 *
 */
public class CustomInterceptor implements Interceptor {
    @Override
    public void initialize() {
        //不需要结合其他的组件，这里什么都不写
    }

    /*
        将接收到的event数据进行解析，重新设置headers或者数据，返回一个新的event
        headers的主要作用是为了后面判断给哪一个sink处理的依据
        判断接收到的数据中是否包含shujia,如果包含，就设置一个键值对在headers中type=sj
        否则设置为type=nsj
     */
    @Override
    public Event intercept(Event event) {
        //获取event中的body内容，body就是我们所监控到的数据
        String info = new String(event.getBody());
        //获取headers,默认情况下，event中的headers是{}
        Map<String, String> headers = event.getHeaders();
        if(info.contains("shujia")){
            headers.put("type","sj");
            event.setHeaders(headers);
        }else {
            headers.put("type","nsj");
            event.setHeaders(headers);
        }

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {

        ArrayList<Event> events1 = new ArrayList<>();

        for (Event event : events) {
            events1.add(intercept(event));
        }
        return events1;
    }

    @Override
    public void close() {
        //因为没有额外的初始化连接，也不需要关闭
    }

    public static class MyBuilder implements Builder{
        @Override
        public Interceptor build() {
            return new CustomInterceptor();
        }

        @Override
        public void configure(Context context) {
        }
    }
}
